package org.lsda.exec;

import java.io.*;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Date;

import javax.jms.Connection;
import javax.jms.Destination;
import javax.jms.ExceptionListener;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.ObjectMessage;
import javax.jms.Session;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.lsda.job.ActionHandler;
import org.lsda.job.LsdaAction;
import org.lsda.util.*;

import test.Action;
import test.Job;

import java.net.*;
import java.io.*;

/**
 * This LsdaWorker class is used to execute the job action. Any node can be 
 * deployed multiple wokers to execute the several jobs at the same moment.
 * And if any node can be added to the job executing cluster by deploying one
 * or more workers.
 * @author grand 
 * */
public class LsdaWorker implements MessageListener, ExceptionListener {
	
	/**the handler of log for this class**/
	private Log log = LogFactory.getLog(LsdaWorker.class);
	
	/**the queue name for transmiting the message**/
	private String queueName = Util.getBRQueName();
	
	/**the url for message broker**/
	private String url = Util.getBRUrl();
	
	/**the user name for connecting the message broker**/
	private String userName = Util.getBRUser();
	
	/**the password for connection the message broker**/
	private String password = Util.getBRPasswd();
	
	/**the refernce of session**/
	private Session session = null;
	
	/**a message producer for reply to the job execution caller**/
	private MessageProducer replyProducer = null;
	
	/**the host name of the current worker. And the default value is local host**/
	private String hostName = "localhost";
	
	/**the list of action handlers**/
	private ArrayList<ActionHandler> actionHandlers = new ArrayList<ActionHandler>();
	
	/**
	 * constructor
	 * */
	public LsdaWorker() {
		// initialize the host name for the worker
		try {
			InetAddress addr = InetAddress.getLocalHost();
			byte[] ipAddr = addr.getAddress();
			hostName = addr.getHostName();
		} catch (UnknownHostException ex) {
			log.error(ex.getMessage());
		}
	}

	/**
	 * get the host name 
	 * */
	public String getHostName() {
		return hostName;
	}
	
	/**
	 * set the host name 
	 * */
	public void setHostName(String hostName) {
		this.hostName = hostName;
	}

	/**
	 * get the name of queue
	 * */
	public String getQueueName() {
		return queueName;
	}

	/**
	 * set the name of queue
	 * */
	public void setQueueName(String queueName) {
		this.queueName = queueName;
	}

	/**
	 * get the url of message broker
	 * */
	public String getUrl() {
		return url;
	}

	/**
	 * set the url of the message broker
	 * */
	public void setUrl(String url) {
		this.url = url;
	}

	/**
	 * get the user name of message broker 
	 * */
	public String getUserName() {
		return userName;
	}
	
	/**
	 * set the user name of message broker
	 * */
	public void setUserName(String userName) {
		this.userName = userName;
	}
	
	/**
	 * get the password of message broker
	 * */
	public String getPassword() {
		return password;
	}

	/**
	 * set the password of message broker
	 * */
	public void setPassword(String password) {
		this.password = password;
	}

	/**
	 * add action handler
	 * */
	public void addActionHandler(ActionHandler ah) {
		actionHandlers.add(ah);
	}

	/**
	 * clear all the action handlers in the list
	 * */
	public void clearActionHandler() {
		actionHandlers.clear();
	}

	/**
	 * worker start to listen
	 * */
	public void jobListen() {
		Connection connection = null;
		try {
			ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
					userName, password, url);
			connection = connectionFactory.createConnection();
			connection.setExceptionListener(this);
			connection.start();

			session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);

			Destination destination = session.createQueue(queueName);

			MessageConsumer consumer = session.createConsumer(destination);
			consumer.setMessageListener(this);
		} catch (Exception ex) {
			ex.printStackTrace();
		}
	}


	/**
	 * when the exception occurs, the method would be called
	 * */
	@Override
	public void onException(JMSException arg0) {
		// TODO Auto-generated method stub
		log.error(arg0.getMessage());
		System.exit(-1);
	}

	/**
	 * after the message arrives, this method will call all the handlers to handle the action
	 * */
	public Action doAct(Action act) {
		String cmd = act.getPrgName() + " " + act.getParam();
		log.info(cmd);
		StringBuffer sbinfo = new StringBuffer();
		String line = null;
		int status = -1;
		try {
			Runtime rt = Runtime.getRuntime();
			Process proc = rt.exec(cmd);
			BufferedReader berr = new BufferedReader(new InputStreamReader(
					proc.getErrorStream()));
			BufferedReader bout = new BufferedReader(new InputStreamReader(
					proc.getInputStream()));
			while ((line = berr.readLine()) != null) {
//				sbinfo.append("stderr:");
				sbinfo.append(line);
				sbinfo.append("\n");
			}

			berr.close();

			while ((line = bout.readLine()) != null) {
//				sbinfo.append("stdout:");
				sbinfo.append(line);
				sbinfo.append("\n");
			}

			bout.close();

			status = proc.waitFor();
			// make sure that the status should be non-negative
			status = Math.abs(status);

		} catch (Exception ex) {
			sbinfo.append(ex.getMessage());
			log.error(ex.getMessage());
			status = 2;
		}
		act.setStatus(status);
		act.setEndTime(new Date());
		act.setInfo(sbinfo.toString());
		log.info(act.getInfo());
		return act;
	}
	
	private LsdaAction doAction(LsdaAction action) {
		String cmd = action.getPrgName() + " " + action.getPrgParam();
		log.info(cmd);
		StringBuffer sbinfo = new StringBuffer();
		String line = null;
		int status = -1;
		int pid = -1;
		String info = "";
		try {
			Runtime rt = Runtime.getRuntime();
			Process proc = rt.exec(cmd);
			BufferedReader berr = new BufferedReader(new InputStreamReader(
					proc.getErrorStream()));
			BufferedReader bout = new BufferedReader(new InputStreamReader(
					proc.getInputStream()));
			while((line = berr.readLine())!=null){
//				sbinfo.append("stderr:");
				sbinfo.append(line);
				sbinfo.append("\n");
			}
			berr.close();
			
			while((line=bout.readLine())!=null){
//				sbinfo.append("stdout:");
				sbinfo.append(line);
				sbinfo.append("\n");
			}
			
			bout.close();
			
			
			status = proc.waitFor();
			
			sbinfo.append("[error code:"+status+"]");
			
		} catch (Exception ex) {
			sbinfo.append( ex.getMessage());
			log.error(ex.getMessage());
			status = 2;
		}
		
		action.setHost(hostName);
		action.setEndTime(new Date());
		action.setInfo(sbinfo.toString());
		action.setStatus(status);
		return action;
	}

	/**
	 * when a message arrives,this method will be called
	 * */
	@Override
	public void onMessage(Message arg0) {
		// TODO Auto-generated method stub
		if (arg0 instanceof ObjectMessage) {
			try {
				ObjectMessage om = (ObjectMessage) arg0;

				LsdaAction action = (LsdaAction) om.getObject();
				replyProducer = session.createProducer(om.getJMSReplyTo());
				Util.markActionRunning(action.getSchedSeq(),action.getActionSeq(), hostName, -1);
				LsdaAction rsp_action = doAction(action);
				ObjectMessage objMsg = session.createObjectMessage(rsp_action);
				objMsg.setJMSCorrelationID("" + rsp_action.getSchedSeq());
				replyProducer.send(objMsg);
				session.commit();
				replyProducer.close();

			} catch (JMSException e) {
				// TODO Auto-generated catch block
				log.error(e.getMessage());
			} catch (ClassNotFoundException e) {
				// TODO Auto-generated catch block
				log.error(e.getMessage());
			} catch (SQLException e) {
				// TODO Auto-generated catch block
				log.error(e.getMessage());
			}
		}
	}

	/**
	 * the worker start to listen for new job
	 * */
	public static void main(String args[]) {
		LsdaWorker worker = new LsdaWorker();
		worker.jobListen();
	}

}
